package com.platform.apex.output;

import java.io.IOException;
import java.net.InetSocketAddress;

import javax.validation.constraints.NotNull;

import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;

import com.datatorrent.lib.db.Connectable;

/**
 * Elastic search base connector which has basic information for an operator
 * <br>
 * Properties:<br>
 * <b>hostname</b>:the host name of the elastic search cluster node to connect
 * to, not null<br>
 * <b>port</b>:port number of the elastic search cluster node to connect to, not
 * null<br>
 * <b>client</b>:created when connected to elastic search cluster node <br>
 * Compile time checks:<br>
 * None<br>
 * <br>
 * Run time checks:<br>
 * hostName batchSize <br>
 * <b>data type:</br>
 * the insertion data can support all the Objects mongoDB supports<br>
 *
 * <b>Benchmarks</b>: <br>
 *
 * @since 2.1.0
 */

public class ElasticAppSearchConnectable implements Connectable {

	@NotNull
	protected String hostName = "";

	@NotNull
	protected int port;

	@NotNull
	protected String clusterName = "elasticsearch";

	protected transient TransportClient client;

	/*
	 * (non-Javadoc)
	 *
	 * @see com.datatorrent.lib.db.Connectable#connect()
	 */
	@Override
	public void connect() throws IOException {
		Settings settings = Settings.settingsBuilder().put("cluster.name", clusterName)
				.put("client.transport.sniff", true).build();
		client = TransportClient.builder().settings(settings).build();
		client.addTransportAddress(new InetSocketTransportAddress(new InetSocketAddress(hostName, port)));
	}

	/*
	 * (non-Javadoc)
	 *
	 * @see com.datatorrent.lib.db.Connectable#disconnect()
	 */
	@Override
	public void disconnect() throws IOException {
		if (client != null) {
			client.close();
		}
	}

	/*
	 * (non-Javadoc)
	 *
	 * @see com.datatorrent.lib.db.Connectable#isConnected()
	 */
	@Override
	public boolean isConnected() {
		if (client != null) {
			return client.connectedNodes().size() != 0;
		}
		return false;
	}

	public String getHostName() {
		return hostName;
	}

	public void setHostName(String hostname) {
		this.hostName = hostname;
	}

	public int getPort() {
		return port;
	}

	public void setPort(int port) {
		this.port = port;
	}

	public String getClusterName() {
		return clusterName;
	}

	public void setClusterName(String clusterName) {
		this.clusterName = clusterName;
	}

}
